package com.zy.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.zy.gmall.realtime.app.func.BeanToJsonStrFunction;
import com.zy.gmall.realtime.app.func.DimAsyncFunction;
import com.zy.gmall.realtime.bean.TradeProvinceOrderBean;
import com.zy.gmall.realtime.util.DateFormatUtil;
import com.zy.gmall.realtime.util.DorisUtil;
import com.zy.gmall.realtime.util.KafkaUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;
import java.util.Collections;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;

public class DwsTradeProvinceOrderWindow {
    //从 Kafka 读取订单明细数据，
    // 过滤 null 数据并按照唯一键对数据去重
    // ，统计各省份各窗口订单数和订单金额，
    // 将数据写入Doris 交易域省份粒度下单各窗口汇总表。
    public static void main(String[] args) {
        //1 创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        //2 设置保存点   开启 超时 重启 两检查点间距 状态后端 是否保存 设置用户
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        //3 获取数据源
        String topic = "dwd_trade_order_detail";
        String groupId = "DwsTradeProvinceOrderWindowGroup";
        KafkaSource<String> kafkaSource = KafkaUtil.getKafkaSource(topic, groupId);
        DataStreamSource<String> sourceDs = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka_source");
        //4 string转为json，过滤空值
        SingleOutputStreamOperator<JSONObject> transDs = sourceDs.process(
                new ProcessFunction<String, JSONObject>() {
                    @Override
                    public void processElement(String str, ProcessFunction<String, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {
                        if (StringUtils.isNotEmpty(str)) {
                            JSONObject jsonObj = JSON.parseObject(str);
                            collector.collect(jsonObj);
                        }
                    }
                }
        );
        //transDs.print(">>>>>>>>");
        //5 按照id分组
        KeyedStream<JSONObject, String> keyByDs = transDs.keyBy(new KeySelector<JSONObject, String>() {
            @Override
            public String getKey(JSONObject jsonObject) throws Exception {
                return jsonObject.getString("id");
            }
        });
        //6 去重 状态 + 消除
        SingleOutputStreamOperator<JSONObject> processDs = keyByDs.process(new KeyedProcessFunction<String, JSONObject, JSONObject>() {
            private ValueState<JSONObject> valueState;

            @Override
            public void open(Configuration parameters) throws Exception {
                ValueStateDescriptor<JSONObject> valueStateDescriptor = new ValueStateDescriptor<JSONObject>("valueStateDescriptor", JSONObject.class);
                valueState = getRuntimeContext().getState(valueStateDescriptor);
            }

            @Override
            public void processElement(JSONObject jsonObj, KeyedProcessFunction<String, JSONObject, JSONObject>.Context context, Collector<JSONObject> out) throws Exception {
                JSONObject lastValue = valueState.value();
                //判断是否存在
                if (lastValue != null) {
                    //不为空，取反
                    String splitTotalAmount = lastValue.getString("split_total_amount");
                    lastValue.put("split_total_amount", "-" + splitTotalAmount);
                    out.collect(lastValue);
                }
                out.collect(jsonObj);
                valueState.update(jsonObj);
            }
        });
        //processDs.print(">>>>>>>");
        //7 转换成实体类
        SingleOutputStreamOperator<TradeProvinceOrderBean> mapDs = processDs.map(new MapFunction<JSONObject, TradeProvinceOrderBean>() {
            @Override
            public TradeProvinceOrderBean map(JSONObject jsonObj) throws Exception {
                String provinceId = jsonObj.getString("province_id");
                String orderId = jsonObj.getString("order_id");
                String splitTotalAmount = jsonObj.getString("split_total_amount");
                Long ts = jsonObj.getLong("ts") * 1000;
                return TradeProvinceOrderBean.builder()
                        .provinceId(provinceId)
                        .orderIdSet(new HashSet(Collections.singleton(orderId)))
                        .orderAmount(new BigDecimal(splitTotalAmount))
                        .ts(ts)
                        .build();
            }
        });
        //mapDs.print(">>>>>>>");

        //8 指定水位线
        SingleOutputStreamOperator<TradeProvinceOrderBean> watermarksDs = mapDs.assignTimestampsAndWatermarks(
                WatermarkStrategy.<TradeProvinceOrderBean>forMonotonousTimestamps()
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<TradeProvinceOrderBean>() {
                                    @Override
                                    public long extractTimestamp(TradeProvinceOrderBean tradeProvinceOrderBean, long l) {
                                        return tradeProvinceOrderBean.getTs();
                                    }
                                }
                        )
        );
        //9 分组
        KeyedStream<TradeProvinceOrderBean, String> keyedStreamDs = watermarksDs.keyBy(new KeySelector<TradeProvinceOrderBean, String>() {
            @Override
            public String getKey(TradeProvinceOrderBean tradeProvinceOrderBean) throws Exception {
                return tradeProvinceOrderBean.getProvinceId();
            }
        });
        //10 开窗
        WindowedStream<TradeProvinceOrderBean, String, TimeWindow> windowDs = keyedStreamDs.window(TumblingEventTimeWindows.of(Time.seconds(5)));
        //11 聚合
        SingleOutputStreamOperator<TradeProvinceOrderBean> reduceDs = windowDs.reduce(
                new ReduceFunction<TradeProvinceOrderBean>() {
                    @Override
                    public TradeProvinceOrderBean reduce(TradeProvinceOrderBean t1, TradeProvinceOrderBean t2) throws Exception {
                        t1.setOrderAmount(t1.getOrderAmount().add(t2.getOrderAmount()));
                        t1.getOrderIdSet().addAll(t2.getOrderIdSet());
                        return t1;
                    }
                },
                new WindowFunction<TradeProvinceOrderBean, TradeProvinceOrderBean, String, TimeWindow>() {
                    @Override
                    public void apply(String str, TimeWindow timeWindow, Iterable<TradeProvinceOrderBean> iterable, Collector<TradeProvinceOrderBean> collector) throws Exception {
                        long start = timeWindow.getStart();
                        String stt = DateFormatUtil.toYmdHms(start);
                        String cur = DateFormatUtil.toDate(start);
                        long end = timeWindow.getEnd();
                        String edt = DateFormatUtil.toYmdHms(end);
                        for (TradeProvinceOrderBean orderBean : iterable) {
                            orderBean.setStt(stt);
                            orderBean.setEdt(edt);
                            orderBean.setCurDate(cur);
                            orderBean.setOrderCount((long) orderBean.getOrderIdSet().size());
                            collector.collect(orderBean);
                        }
                    }
                }
        );
        //reduceDs.print(">>>>>>>");
        //12 维度关联 使用IO异步
        SingleOutputStreamOperator<TradeProvinceOrderBean> nameDs = AsyncDataStream.unorderedWait(
                reduceDs,
                new DimAsyncFunction<TradeProvinceOrderBean>("dim_base_province") {
                    @Override
                    public void join(TradeProvinceOrderBean obj, JSONObject jsonObj) {
                        //连接一下
                        obj.setProvinceName(jsonObj.getString("name"));
                    }

                    @Override
                    public String getRowKey(TradeProvinceOrderBean obj) {
                        return obj.getProvinceId();
                    }
                },
                60,
                TimeUnit.SECONDS
        );
        // nameDs.print(">>>>>>>>");
        //13 写入doris
        nameDs.print(">>>>>>>");
        nameDs
                .map(new BeanToJsonStrFunction<>())
                .sinkTo(DorisUtil.getDorisSink("dws_trade_province_order_window"));
        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
//stream load error: [MEM_LIMIT_EXCEEDED]Process has no memory available,
// cancel top memory usage load: load memory tracker <Load#Id=a94d720a1d94d3b1-f3cb346708e14486> consumption 1.25 MB,
// backend 192.168.10.103 process memory used 343.12 MB exceed limit 2.31 GB or sys mem available 56.25 MB less than low water mark 259.16 MB.
// Execute again after enough memory, details see be.INFO., see more in null
